package com.example.latch;

import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.grpc.stub.StreamObserver;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.function.BiConsumer;


/***
 * Etcd实现选主策列
 */
@Data
public class EtcdLeaderLatch implements EtcdLatch {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Client client;

    private String lockKey;

    private String nodeKey;

    private ByteSequence lockSequence;

    private ByteSequence nodeSequence;

    private long ttl;

    private long grantId;

    private EtcdLeaderLatchListener listener;

    private Thread thread;

    public EtcdLeaderLatch(Client client, String lockKey, String nodeKey, long ttl, EtcdLeaderLatchListener listener) {
        this.client = client;
        this.lockKey = lockKey;
        this.nodeKey = nodeKey;
        this.ttl = ttl;
        this.listener = listener;
        this.lockSequence = ByteSequence.from(lockKey.getBytes(StandardCharsets.UTF_8));
        this.nodeSequence = ByteSequence.from(nodeKey.getBytes(StandardCharsets.UTF_8));
    }

    public EtcdLeaderLatch(Client client, String lockKey, String nodeKey, long ttl, EtcdLeaderLatchListener listener, StreamObserver<LeaseKeepAliveResponse> streamObserver) {
        this.client = client;
        this.lockKey = lockKey;
        this.nodeKey = nodeKey;
        this.ttl = ttl;
        this.listener = listener;
        this.lockSequence = ByteSequence.from(lockKey.getBytes(StandardCharsets.UTF_8));
        this.nodeSequence = ByteSequence.from(nodeKey.getBytes(StandardCharsets.UTF_8));
        this.streamObserver = streamObserver;
    }

    private StreamObserver<LeaseKeepAliveResponse> streamObserver = new StreamObserver<LeaseKeepAliveResponse>() {

        @Override
        public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {

        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onCompleted() {

        }
    };

    @Override
    public void start() {
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                client.getLeaseClient().grant(ttl).whenComplete(new BiConsumer<LeaseGrantResponse, Object>() {
                    @Override
                    public void accept(LeaseGrantResponse leaseGrantResponse, Object o) {
                        grantId = leaseGrantResponse.getID();
                        client.getLeaseClient().keepAlive(grantId, streamObserver);
                        client.getElectionClient().campaign(lockSequence, grantId, nodeSequence).whenComplete(new BiConsumer() {
                            @Override
                            public void accept(Object o, Object o2) {
                                listener.isLeader();
                            }
                        });
                    }
                });
            }
        });
        this.thread.setName("EtcdMasterThread-" + lockKey);
        this.thread.setDaemon(true);
        this.thread.start();
    }

    @Override
    public void stop() {
        this.client.getLeaseClient().revoke(grantId);
        if (this.thread.isAlive()) {
            this.thread.interrupt();
        }
    }
}
